package org.zhadoop.stream.cql.mapping;

import java.util.Map;
import java.util.Map.Entry;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.zhadoop.stream.api.opereators.serdes.CSVSerDeAPI;
import org.zhadoop.stream.api.opereators.serdes.SimpleSerDeAPI;
import org.zhadoop.stream.cql.executor.operatorinfocreater.KafkaInputOperator;
import org.zhadoop.streaming.operator.inputstream.KafkaSourceOp;
import org.zhadoop.streaming.serde.CSVSerDe;
import org.zhadoop.streaming.serde.SimpleSerDe;

import com.google.common.collect.Maps;

/**
 * 输入或者输出算子和API之间的映射关系
 * @author zwd
 *
 */
public class InputOutputOperatorMapping {
	private static final Logger LOG = LoggerFactory.getLogger(InputOutputOperatorMapping.class); 
	
	private static final Map<String,String> API_PLAT_MAPPING = Maps.newConcurrentMap();
		
	static{
		/**
		 * 输入输出算子
		 */
		API_PLAT_MAPPING.put(KafkaInputOperator.class.getName(), KafkaSourceOp.class.getName());
		
		/**
		 * 序列化，反序列化类
		 */
		API_PLAT_MAPPING.put(SimpleSerDeAPI.class.getName(), SimpleSerDe.class.getName());
		API_PLAT_MAPPING.put(CSVSerDeAPI.class.getName(), CSVSerDe.class.getName());
	}

	/**
	 * 底层的算子名称或者对应的API类
	 * @param deserClassName
	 * @return
	 */
	public static String getAPIOperatorByPlatform(String platformClassName) {
		for(Entry<String,String> et : API_PLAT_MAPPING.entrySet()){
			if(et.getValue().equals(platformClassName)){
				return et.getKey();
			}
		}
		return null;
	}
}
